package com.shujia.flink.source;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.core.fs.Path;
import org.apache.flink.formats.csv.CsvReaderFormat;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.dataformat.csv.CsvMapper;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.dataformat.csv.CsvSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;

/**
 * @author SHUJIA
 */
public class Demo6StudentRead {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();


        CsvMapper csvMapper = new CsvMapper();

        CsvSchema csvSchema = csvMapper.schemaFor(Students.class)
                //增加字段顺序
                .sortedBy("id", "name", "age", "gender", "clazz")
                //增加字段分隔符
                .withColumnSeparator('|');

        //定义读取文件的格式
        CsvReaderFormat<Students> csvReaderFormat = CsvReaderFormat
                .forSchema(csvSchema, TypeInformation.of(Students.class));

        //file source
        FileSource<Students> fileSource = FileSource
                //指定读取数据的格式和路径
                .forRecordStreamFormat(csvReaderFormat, new Path("data/student"))
                //指定循环监控目录的间隔时间
                .monitorContinuously(Duration.ofSeconds(1))
                .build();

        //使用file source
        DataStream<Students> student = env
                .fromSource(fileSource, WatermarkStrategy.noWatermarks(), "studentSource");

        student.print();
        env.execute();
    }
}
